package com.fwmagic.flink.datasource;

import com.fwmagic.flink.pojo.Student;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class SourceFromMysql extends RichSourceFunction<Student> {

    private Connection connection;

    private PreparedStatement ps;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();
        String sql ="select * from student";
        ps = this.connection.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        ResultSet rs = ps.executeQuery();
        while(rs.next()){
            Student student = new Student();
            student.setName(rs.getString("name"));
            student.setAge(rs.getInt("age"));
            student.setId(rs.getInt("id"));
            student.setPassword(rs.getString("password"));
            ctx.collect(student);
        }
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        super.close();
        if (ps != null) ps.close();
        if (connection != null) connection.close();
    }

    private Connection getConnection() {
        Connection connection = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "123456");
        } catch (Exception e) {
            e.printStackTrace();
            System.err.println("====>get connection exception!" + e.getLocalizedMessage());
        }
        return connection;
    }


}
